package cn.hy.dataSync.task;


import cn.hy.dataSync.job.State;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 任务监控器
 */
@Slf4j
public class TaskMonitor {

    private static final TaskMonitor INSTANCE = new TaskMonitor();

    /**
     * 2天（毫秒）
     */
    private static final long EXPIRED_TIME = 172800 * 1000;

    private ConcurrentHashMap<Integer, TaskCommunication> tasks = new ConcurrentHashMap<>();

    private TaskMonitor() {
    }

    public static TaskMonitor getInstance() {
        return INSTANCE;
    }

    public void registerTask(Integer taskId, Communication communication) {
        //如果task已经finish，直接返回
        if (communication.isFinished()) {
            return;
        }
        tasks.putIfAbsent(taskId, new TaskCommunication(taskId, communication));
    }

    public void removeTask(Integer taskId) {
        tasks.remove(taskId);
    }

    public void report(Integer taskId, Communication communication) {
        //如果task已经finish，直接返回
        if (communication.isFinished()) {
            return;
        }
        if (!tasks.containsKey(taskId)) {
            log.warn("unexpected: taskId({}) missed.", taskId);
            tasks.putIfAbsent(taskId, new TaskCommunication(taskId, communication));
        } else {
            tasks.get(taskId).report(communication);
        }
    }

    public TaskCommunication getTaskCommunication(Integer taskId) {
        return tasks.get(taskId);
    }

    public static class TaskCommunication {
        private Integer taskId;
        //记录最后更新的communication
        private long lastAllReadRecords;
        //只有第一次，或者统计变更时才会更新
        private long lastUpdateCommunicationTs;
        private long ttl;

        private TaskCommunication(Integer taskId, Communication communication) {
            this.taskId = taskId;
            lastAllReadRecords = CommunicationTool.getTotalReadRecords(communication);
            ttl = System.currentTimeMillis();
            lastUpdateCommunicationTs = ttl;
        }

        public void report(Communication communication) {
            ttl = System.currentTimeMillis();
            if (CommunicationTool.getTotalReadRecords(communication) > lastAllReadRecords) {
                lastAllReadRecords = CommunicationTool.getTotalReadRecords(communication);
                lastUpdateCommunicationTs = ttl;
            } else if (isExpired(lastUpdateCommunicationTs)) {
                communication.setState(State.FAILED);
                communication.setTimestamp(ttl);
                communication.setThrowable(new RuntimeException(String.format("task(%s) hung expired [allReadRecord(%d), elased(%s)]",
                        taskId, lastAllReadRecords, (ttl - lastUpdateCommunicationTs))));
            }
        }

        private boolean isExpired(long lastUpdateCommunicationTs) {
            return System.currentTimeMillis() - lastUpdateCommunicationTs > EXPIRED_TIME;
        }

        public Integer getTaskId() {
            return taskId;
        }

        public long getLastAllReadRecords() {
            return lastAllReadRecords;
        }

        public long getLastUpdateCommunicationTs() {
            return lastUpdateCommunicationTs;
        }

        public long getTtl() {
            return ttl;
        }
    }
}
